#include "raft_core/RaftCore.h"

#include <async_simple/coro/Sleep.h>
#include <grpcpp/support/status.h>

#include <cassert>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include "raft_core/ApplyMsg.h"
#include "raft_core/RaftSender.h"
#include "raft_core/rpc_proto/raft_rpc.pb.h"
#include "raft_utils/Command.h"
#include "raft_utils/Config.h"
#include "raft_utils/DataStream.h"
#include "raft_utils/Persistence.h"


raft::core::RaftCore::RaftCore(std::shared_ptr<state_machine::StateMachine> state_machine,
                               std::shared_ptr<utils::Command> command)
    : state_machine_(std::move(state_machine)),
      command_(std::move(command)),
      status_(Status::Follower),
      executor_(2),
      log(bq::log::get_log_by_name(utils::Config::getInstance()->get_log_name())) {

    auto config = utils::Config::getInstance();

    auto peers = config->get_raft_core_peer();
    for (const auto& i: peers) {
        auto raft_sender = std::make_unique<RaftSender>(i.first, i.second);
        peers_.emplace_back(std::move(raft_sender));
    }

    current_term_ = 0;
    voted_for_.clear();
    logs_.clear();
    last_snapshot_include_index_ = 0;
    last_snapshot_include_term_ = 0;
    commit_index_ = 0;
    last_applied_ = 0;
    leader_me_.clear();

    // 遍历所有服务器，初始化匹配索引数组
    for (int i = 0; i < peers_.size(); ++i) {
        match_index_.emplace_back(0);
        next_index_.emplace_back(0);
    }

    last_election_reset_time_ = std::chrono::high_resolution_clock::now();
    last_heartbeat_reset_time_ = std::chrono::high_resolution_clock::now();

    me_ = config->get_raft_core_ip() + ":" + std::to_string(config->get_raft_core_port());
    persistence_ = std::make_unique<utils::Persistence>(me_);

    apply_msg_ = std::make_unique<ApplyMsg>();

    // 从快照中恢复
    read_persistence(persistence_->read_raft_state());
    // 如果快照的日志索引大于0，则说明有相关有效数据
    if (last_snapshot_include_index_ > 0) {
        last_applied_ = last_snapshot_include_index_;
    }

    log.info("[func-raft::core::RaftCore::RaftCore] Server: {}, term: {}, last_snapshot_include_index: {}, last_snapshot_include_term: {}",
             me_, current_term_, last_snapshot_include_index_, last_snapshot_include_term_);

    // 启动协程，持续心跳运行心跳，选举检查等操作
    this->leader_heartbeat_timer().via(&executor_).start([](auto&&) {});
    this->election_timer().via(&executor_).start([](auto&&) {});

    // 启动一个线程执行将日志应用到状态机操作
    std::thread t(&RaftCore::find_apply_log_timer, this);
    t.detach();
}


/* ------ 序列化相关 ------ */
std::string raft::core::RaftCore::serialize_data() {
    log.verbose("Start: std::string raft::core::RaftCore::serialize_data()");
    utils::DataStream data_stream;
    data_stream << current_term_ << voted_for_ << last_snapshot_include_index_
                << last_snapshot_include_term_;
    std::vector<std::string> temp_logs;
    temp_logs.reserve(logs_.size());
    for (const auto& i: logs_) {
        temp_logs.emplace_back(i.SerializeAsString());
    }
    data_stream << temp_logs;
    std::string serialize_str(data_stream.data(), data_stream.size());

    log.verbose("End: std::string raft::core::RaftCore::serialize_data()");
    return serialize_str;
}

void raft::core::RaftCore::persistence() {
    log.verbose("Start: void raft::core::RaftCore::persistence()");
    auto data = serialize_data();
    persistence_->save_raft_state(data);
    log.verbose("End: void raft::core::RaftCore::persistence()");
}

void raft::core::RaftCore::read_persistence(const std::string& data) {
    log.verbose("Start: void raft::core::RaftCore::read_persistence(const std::string& data)");

    if (data.empty()) {
        log.verbose("End: void raft::core::RaftCore::read_persistence(const std::string& data)");
        return;
    }
    utils::DataStream data_stream;
    std::vector<char> serialize_buffer(data.c_str(), data.c_str() + data.size());
    data_stream.write(serialize_buffer.data(), serialize_buffer.size());

    {
        std::lock_guard<std::mutex> lock(mutex_);
        data_stream >> current_term_ >> voted_for_ >> last_snapshot_include_index_ >> last_snapshot_include_term_;
        std::vector<std::string> temp_logs;
        data_stream >> temp_logs;
        for (auto& i: temp_logs) {
            raft_rpc_protoc::LogEntry log_entry;
            log_entry.ParseFromString(i);
            logs_.emplace_back(log_entry);
        }
    }

    log.verbose("End: void raft::core::RaftCore::read_persistence(const std::string& data)");
}

/* ------ 快照相关 ------ */
long long raft::core::RaftCore::get_raft_state_size() {
    log.verbose("Start: int raft::core::RaftCore::get_raft_state_size()");
    log.verbose("End: int raft::core::RaftCore::get_raft_state_size()");
    return persistence_->get_raft_state_size();
}

void raft::core::RaftCore::save_snapshot(int index, const std::string& state_machine_snapshot) {
    log.verbose("Start: void raft::core::RaftCore::save_snapshot(int index)");

    // 检查传入的快照索引是否有效
    if (index < last_snapshot_include_index_ || index > commit_index_) {
        log.verbose("End: void raft::core::RaftCore::save_snapshot(int index)");
        return;
    }

    // 获取当前最后一个日志作为索引
    auto last_log_index = get_last_log_index();

    // 设置新快照的所有和任期
    int new_last_snapshot_include_index = index;
    int new_last_snapshot_include_term = logs_[get_slices_index_by_log_index(index)].log_term();

    // 截断日志，只保留大于 index 的日志条目
    std::vector<raft_rpc_protoc::LogEntry> truncked_logs;
    for (int i = index + 1; i <= get_last_log_index(); ++i) {
        truncked_logs.emplace_back(logs_[get_slices_index_by_log_index(i)]);
    }

    {
        std::lock_guard<std::mutex> lock(mutex_);
        // 更新节点状态
        last_snapshot_include_index_ = new_last_snapshot_include_index;
        last_snapshot_include_index_ = new_last_snapshot_include_term;
        logs_ = truncked_logs;
        commit_index_ = std::max(commit_index_, index);
        last_applied_ = std::max(last_applied_, index);
    }

    // 将快照数据持久化保存
    persistence_->save(serialize_data(), state_machine_snapshot);

    log.verbose("End: void raft::core::RaftCore::save_snapshot(int index)");
}

void raft::core::RaftCore::restore_state_machine_snapshot(const std::string& snapshot) {
    log.verbose("Start: void raft::core::RaftCore::restore_state_machine_snapshot(const std::string& snapshot)");

    state_machine_->restore(snapshot);

    log.verbose("End: void raft::core::RaftCore::restore_state_machine_snapshot(const std::string& snapshot)");
}


/* ------ 日志相关 ------ */
int raft::core::RaftCore::get_last_log_index() {
    log.verbose("Start: int raft::core::RaftCore::get_last_log_index()");

    if (logs_.empty()) {
        log.verbose("End: int raft::core::RaftCore::get_last_log_index()");
        return last_snapshot_include_index_;
    }
    log.verbose("End: int raft::core::RaftCore::get_last_log_index()");
    return logs_[logs_.size() - 1].log_index();
}

int raft::core::RaftCore::get_last_log_term() {
    log.verbose("Start: int raft::core::RaftCore::get_last_log_term()");

    if (logs_.empty()) {
        log.verbose("End: int raft::core::RaftCore::get_last_log_term()");
        return last_snapshot_include_term_;
    }
    log.verbose("End: int raft::core::RaftCore::get_last_log_term()");
    return logs_[logs_.size() - 1].log_term();
}

bool raft::core::RaftCore::is_latest(int index, int term) {
    log.verbose("Start: bool raft::core::RaftCore::is_latest(int index, int term)");

    auto last_log_index = get_last_log_index();
    auto last_log_term = get_last_log_term();

    log.verbose("End: bool raft::core::RaftCore::is_latest(int index, int term)");
    return (term > last_log_term) || (term == last_log_term && index >= last_log_index);
}

// 将日志索引转换为日志条目在logs_数组中的位置
int raft::core::RaftCore::get_slices_index_by_log_index(int log_index) {
    log.verbose("Start: int raft::core::RaftCore::get_slices_index_by_log_index(int log_index)");
    if (log_index < last_snapshot_include_index_) {
        log.fatal("[func-raft_raft_core::Raft::get_slices_index_by_log_index] log_index < last_snapshot_include_index_",
                  log_index, last_snapshot_include_index_);
        return -1;
    }
    log.verbose("End: int raft::core::RaftCore::get_slices_index_by_log_index(int log_index)");
    return log_index - last_snapshot_include_index_ - 1;
}

void raft::core::RaftCore::get_prev_log_info(int server, int* pre_index, int* pre_term) {
    log.verbose("Start: void raft::core::RaftCore::get_prev_log_info(int server, int* pre_index, int* pre_term)");

    // 如果下一个索引是快照包含的索引 + 1，说明要发送的日志是第一个日志
    if (next_index_[server] == last_snapshot_include_index_ + 1) {
        *pre_index = last_snapshot_include_index_;
        *pre_term = last_snapshot_include_term_;
        log.verbose("End: void raft::core::RaftCore::get_prev_log_info(int server, int* pre_index, int* pre_term)");
        return;
    }
    // 否则，计算前一个日志条目索引
    *pre_index = next_index_[server] - 1;
    *pre_term = logs_[get_slices_index_by_log_index(*pre_index)].log_term();

    log.verbose("End: void raft::core::RaftCore::get_prev_log_info(int server, int* pre_index, int* pre_term)");
}

int raft::core::RaftCore::get_log_term_by_log_index(int log_index) {
    log.verbose("Start: int raft_core::Raft::get_log_term_by_log_index(int log_index)");
    if (log_index < last_snapshot_include_index_) {
        log.fatal("[func-raft_core::Raft::get_log_term_by_log_index] log_index < last_snapshot_include_index_",
                  log_index, last_snapshot_include_index_);
        return -1;
    }
    int last_log_index = get_last_log_index();
    log.verbose("End: raft_core::Raft::get_log_term_by_log_index(int log_index)");
    if (log_index == last_snapshot_include_index_) {
        return last_snapshot_include_term_;
    } else {
        return logs_[get_slices_index_by_log_index(log_index)].log_term();
    }
}

bool raft::core::RaftCore::is_log_match(int log_index, int log_term) {
    log.verbose("Start: bool raft::core::RaftCore::is_log_match(int log_index, int log_term)");

    if (log_index < last_snapshot_include_index_ && log_index > get_last_log_index()) {
        log.fatal("[func-raft::core::RaftCore::is_log_match](int log_index, int log_term)] log_index{} < last_snapshot_include_index_{} && log_index{} > get_last_log_index(){}",
                  log_index, last_snapshot_include_index_, log_index, get_last_log_index());
        return false;
    }

    log.verbose("End: bool raft::core::RaftCore::is_log_match(int log_index, int log_term)");
    return log_term == get_log_term_by_log_index(log_index);
}


/* ------ 状态机相关 ------ */
void raft::core::RaftCore::find_apply_log_timer() {
    log.debug("[func-raft::core::RaftCore::find_apply_log_timer()] Start find apply log timer loop");
    while (true) {
        if (commit_index_ > get_last_log_index()) {
            log.fatal("[func-raft_core::Raft::get_apply_msg()] commit_index_ > get_last_log_index()",
                      commit_index_, get_last_log_index());
        }
        // 获取可以提交给状态机的日志
        std::vector<ApplyMsg> apply_msgs;
        while (last_applied_ < commit_index_) {
            // 设置命令
            ApplyMsg apply_msg;
            apply_msg.set_command_valid(true);
            if (last_applied_ == 0 && last_snapshot_include_index_ == 0) {
                apply_msg.set_command(logs_[0].command());
            } else {
                apply_msg.set_command(logs_[get_slices_index_by_log_index(last_applied_) + 1].command());
            }
            apply_msg.set_commint_index(last_applied_);
            apply_msgs.emplace_back(apply_msg);
            ++last_applied_;
        }

        for (auto& i: apply_msgs) {
            apply_log(i);
        }

        // 应用日志完成后休眠
        std::this_thread::sleep_for(std::chrono::milliseconds(utils::Config::getInstance()->get_apply_interval()));
    }
}

void raft::core::RaftCore::apply_log(ApplyMsg& log_msg) {
    log.verbose("Start: void raft::core::RaftCore::apply_log(ApplyMsg& apply_msg)");

    if (log_msg.is_command_valid()) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            command_->parse_from_string(log_msg.get_command());
            std::string msg;
            auto result = state_machine_->apply(command_, &msg);
            if (!result) {
                log.info("[func-raft::core::RaftCore::apply_log] apply msg: {}", msg);
            }
            command_->clear();
        }
        // 判断是否需要制作快照
        if (get_raft_state_size() > utils::Config::getInstance()->get_max_raft_state()) {
            auto state_machine_snapshot = state_machine_->snapshot();
            save_snapshot(log_msg.get_commint_index(), state_machine_snapshot);
        }
    }

    if (log_msg.is_snapshot_valid()) {
        std::lock_guard<std::mutex> lock(mutex_);
        restore_state_machine_snapshot(log_msg.get_snapshot());
        last_snapshot_include_index_ = log_msg.get_snapshot_index();
        last_snapshot_include_term_ = log_msg.get_snapshot_term();
    }

    log.verbose("End: void raft::core::RaftCore::apply_log(ApplyMsg& apply_msg)");
}


/* ------ 选举相关 ------ */
async_simple::coro::Lazy<void> raft::core::RaftCore::election_timer() {
    log.debug("[func-raft::core::RaftCore::election_timer()] Start election timer loop");
    // 循环检查是否应该开始领导人选举
    while (true) {
        log.debug("[func-raft::core::RaftCore::election_timer()] starting....");
        // 如果当前节点为 leader 则不需要进行选举，进入休眠状态减少 CPU 消耗
        while (status_ == Status::Leader) {
            // 这里使用的心跳超时时间 * 2（心跳超时比选举超时一般小一个数量级）
            co_await async_simple::coro::sleep(std::chrono::milliseconds(utils::Config::getInstance()->get_heart_beat_timeout() * 2));
        }


        // 定义唤醒时间点
        std::chrono::system_clock::time_point wake_time{};
        wake_time = std::chrono::high_resolution_clock::now();

        std::random_device rd;
        std::mt19937 rng(rd());
        std::uniform_int_distribution<int> dist(utils::Config::getInstance()->get_min_randomized_vote_time(),
                                                utils::Config::getInstance()->get_max_randomized_vote_time());

        // 随机化选举超时，为了避免多个节点同时成为 Candidate 并发起选举（称为“选举分裂”），选举超时时间是随机化的
        std::chrono::nanoseconds suitable_sleep_time = std::chrono::milliseconds(dist(rng)) + last_election_reset_time_ - wake_time;

        // 如果计算出来的时间大于1毫秒，则休眠
        if (std::chrono::duration<double, std::milli>(suitable_sleep_time).count() > 1) {
            // 记录开始时间点
            auto start = std::chrono::steady_clock::now();
            // 协程休眠
            co_await async_simple::coro::sleep(suitable_sleep_time);
            // 记录结束时间点
            auto end = std::chrono::steady_clock::now();

            log.debug("[func-raft_core::Raft::vote_timeout_ticker()] The function sets the sleep time to: {}, the actual sleep time of the function is: {}",
                      std::chrono::duration_cast<std::chrono::milliseconds>(suitable_sleep_time).count(),
                      (end - start));
        }

        // 如果休眠期间选举定时器被重置，则不需要选举
        if (last_election_reset_time_ > wake_time) {
            continue;
        }

        // 否则超时，进行发起选举
        initiate_election();
    }
}

void raft::core::RaftCore::initiate_election() {
    log.verbose("Start: void raft::core::RaftCore::initiate_election()");

    // 如果节点为 leader, 不需要选举
    if (status_ == Status::Leader) {
        log.verbose("End: void raft::core::RaftCore::initiate_election()");
        return;
    }

    // 开始选举
    log.info("[func-raft::core::RaftCore::initiate_election()] {} Election timer expires and is not the leader, start the election.",
             me_);
    {
        std::lock_guard<std::mutex> lock(mutex_);
        // 设置状态为 candidate
        status_ = Status::Candidate;
        // 增加当前任期号
        ++current_term_;
        // 给自己投一票，每个 candidate 在开始选举前都会先投自己一票
        voted_for_ = me_;
    }

    // 持久化数据
    persistence();

    {
        std::lock_guard<std::mutex> lock(mutex_);
        // 重置定时器
        last_election_reset_time_ = std::chrono::high_resolution_clock::now();
    }
    // 计算选票数，使用原子计数器的方式
    auto vote_num = std::make_shared<std::atomic<int>>(1);
    // 获取最后一个日志条目索引和任期号
    auto last_log_index = get_last_log_index();
    auto last_log_term = get_last_log_term();


    // 构造 rpc 请求
    auto request = std::make_shared<raft_rpc_protoc::RequestElection>();
    request->set_term(current_term_);
    request->set_candidate_id(me_);
    request->set_last_log_index(last_log_index);
    request->set_last_log_term(last_log_term);

    // 向其他节点发送请求 request_election rpc 请求
    for (int i = 0; i < peers_.size(); ++i) {
        auto reply = std::make_shared<raft_rpc_protoc::RequestElectionReply>();
        // 启动协程发送
        this->send_request_election(i, request, reply, vote_num).start([this](auto&&) {});
    }
    log.verbose("End: void raft::core::RaftCore::initiate_election()");
}

async_simple::coro::Lazy<bool>
raft::core::RaftCore::send_request_election(int server,
                                            std::shared_ptr<raft_rpc_protoc::RequestElection> request,
                                            std::shared_ptr<raft_rpc_protoc::RequestElectionReply> reply,
                                            std::shared_ptr<std::atomic<int>> vote_num) {
    log.verbose("Start: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_request_election(int server,...)");

    auto result = peers_[server]->request_election(*request, reply.get());
    if (!result) {
        log.debug("[func-raft::core::RaftCore::send_request_election] send request_vote failed");
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_request_election(int server,...)");
        co_return false;
    }

    // 检测响应中的任期号，如果大于当前节点的任期号，则更新任期号并转化身份为 follower
    if (reply->term() > current_term_) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = reply->term();
            voted_for_.clear();
            leader_me_ = request->candidate_id();
        }
        // 持久化以上数据
        persistence();
        co_return true;
    } else if (reply->term() < current_term_) {
        // 如果小于当前节点任期号，忽略响应
        co_return true;
    }

    // 如果对方没有投票给当前节点
    if (!reply->vote_granted()) {
        co_return true;
    }

    // 增加票数
    ++*vote_num;
    log.debug("[func-raft::core::RaftCore::send_request_election] vote_num: {}",
              vote_num->load());
    // 如果超过半数，则成为 leader
    if (*vote_num > (peers_.size() / 2)) {
        // 重置计数器
        *vote_num = 0;
        if (status_ == Status::Leader) {
            log.fatal("[func-raft::core::RaftCore::send_request_election] {} term: {}, Same term as two leaders",
                      me_, current_term_);
            co_return false;
        }

        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Leader;
        }

        log.info("[func-raft::core::RaftCore::send_request_election] {} elect success, current_term: {}, last_log_index: {}",
                 me_, current_term_, get_last_log_index());

        // 初始化 next_index 和 match_index;
        for (int i = 0; i < next_index_.size(); ++i) {
            next_index_[i] = get_last_log_index() + 1;
            match_index_[i] = 0;
        }

        // 启动一个协程来发送心跳或复制日志，宣告成为 leader
        this->initiate_heartbeat().start([](auto&&) {});

        // 持久化状态
        persistence();
    }

    log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_request_election(int server,...)");
    co_return true;
}

grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,
                                                    const raft_rpc_protoc::RequestElection* request,
                                                    raft_rpc_protoc::RequestElectionReply* reply) {
    log.verbose("Start: grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,...)");

    log.debug("[func-raft::core::RaftCore::request_election] {} Received election request",
              me_);

    // 检查请求中的任期号
    if (request->term() < current_term_) {
        // 如果请求的任期号小于当前节点的任期号，拒绝投票
        reply->set_term(current_term_);
        reply->set_vote_granted(false);
        reply->set_vote_state(2);
        log.verbose("End: grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    } else if (request->term() > current_term_) {
        // 如果是大于当前节点任期号，更新当前节点的任期号，并转为 follower
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = request->term();
            voted_for_.clear();
            leader_me_ = request->candidate_id();
        }
        persistence();
    }

    if (request->term() != current_term_) {
        log.fatal("[func-raft::core::RaftCore::request_election] {} Previously verified that request.term==Current_term is not equal here.",
                  me_);
        persistence();
        return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, "fatal error");
    }

    // 如果候选人的日志比接收者的日志旧，拒绝投票
    if (!is_latest(request->last_log_index(), request->last_log_term())) {
        reply->set_term(current_term_);
        reply->set_vote_granted(false);
        reply->set_vote_state(2);
        persistence();
        log.verbose("End: grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }
    // 如果已经投票给其他候选人，拒绝投票
    if (!voted_for_.empty() && voted_for_ != request->candidate_id()) {
        reply->set_term(current_term_);
        reply->set_vote_granted(false);
        reply->set_vote_state(2);
        persistence();
        log.verbose("End: grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    {
        std::lock_guard<std::mutex> lock(mutex_);
        // 投票给请求者
        voted_for_ = request->candidate_id();
        leader_me_ = request->candidate_id();
    }

    reply->set_term(current_term_);
    reply->set_vote_granted(true);
    reply->set_vote_state(3);

    persistence();
    {
        std::lock_guard<std::mutex> lock(mutex_);
        // 重置选举定时器
        last_election_reset_time_ = std::chrono::high_resolution_clock::now();
    }

    log.verbose("End: grpc::Status raft::core::RaftCore::request_election(grpc::ServerContext* context,...)");
    return grpc::Status::OK;
}


/* ------ 日志复制与心跳相关 ------ */
async_simple::coro::Lazy<void> raft::core::RaftCore::leader_heartbeat_timer() {
    log.debug("[func-raft::core::RaftCore::leader_heartbeat_timer()] Start leader heartbet timer loop");
    while (true) {
        log.debug("[func-raft::core::RaftCore::leader_heartbeat_timer()] starting...");
        // 如果不是 leader 则进入休眠
        while (status_ != Status::Leader) {
            co_await async_simple::coro::sleep(std::chrono::milliseconds(utils::Config::getInstance()->get_heart_beat_timeout() * 1000));
        }

        // 定义合适的睡眠时间和唤醒时间
        std::chrono::nanoseconds suitable_sleep_time{};
        std::chrono::system_clock::time_point wake_time{};

        wake_time = std::chrono::high_resolution_clock::now();
        // 睡眠时间 = 心跳超时时间 + 上次重置心跳时间 - 当前时间
        // 计算睡眠时间 = 心跳超时时间 - (当前时间 - 上次重置心跳时间)
        suitable_sleep_time = std::chrono::milliseconds(utils::Config::getInstance()->get_heart_beat_timeout()) - (wake_time - last_heartbeat_reset_time_);

        // 如果睡眠时间大于 1 ms，则进入睡眠
        if (suitable_sleep_time > std::chrono::milliseconds(1)) {
            // 记录开始时间点
            auto start = std::chrono::steady_clock::now();
            // 协程休眠
            co_await async_simple::coro::sleep(suitable_sleep_time);
            // 记录结束时间点
            auto end = std::chrono::steady_clock::now();

            log.debug("[func-raft_core::Raft::leader_heartbeat_ticker()] The function sets the sleep time to: {} ms, the actual sleep time of the function is: {} ms",
                      std::chrono::duration_cast<std::chrono::milliseconds>(suitable_sleep_time).count(),
                      std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count());
        }

        // 如果休眠期间心跳被重置，则跳过本次心跳
        if (last_heartbeat_reset_time_ > wake_time) {
            continue;
        }

        // 否则发送心跳或日志复制
        co_await initiate_heartbeat();
    }
}

async_simple::coro::Lazy<void> raft::core::RaftCore::initiate_heartbeat() {
    log.verbose("Start: async_simple::coro::Lazy<void> raft::core::RaftCore::initiate_heartbeat()");

    if (status_ != Status::Leader) {
        log.verbose("End: async_simple::coro::Lazy<void> raft::core::RaftCore::initiate_heartbeat()");
        co_return;
    }

    // 正确返回的节点数量
    auto append_nums = std::make_shared<std::atomic<int>>(1);

    // 向其他节点发送 append_entries rpc 请求
    for (int i = 0; i < peers_.size(); ++i) {

        if (next_index_[i] < 1) {
            log.fatal("raft.next_index{} = {}", i, next_index_[i]);
            co_return;
        }

        // 如果发送给 follower 的下一个节点小于当前快照，则发送快照
        if (next_index_[i] < last_snapshot_include_index_) {
            // 构造 rpc 请求
            auto request = std::make_shared<raft_rpc_protoc::InstallSnapshot>();
            request->set_leader_id(me_);
            request->set_term(current_term_);
            request->set_last_snapshot_include_index(last_snapshot_include_index_);
            request->set_last_snapshot_include_term(last_snapshot_include_term_);
            request->set_data(persistence_->read_snapshot());

            auto reply = std::make_shared<raft_rpc_protoc::InstallSnapshotReply>();

            // 启动一个协程发送
            this->send_install_snapshot(i, request, reply).start([this](auto&&) {});
            continue;
        }

        // 否则发送 append_entries
        int prev_log_index = -1, prev_log_term = -1;
        get_prev_log_info(i, &prev_log_index, &prev_log_term);

        // 构造 rpc 请求
        auto request = std::make_shared<raft_rpc_protoc::AppendEntries>();
        request->set_term(current_term_);
        request->set_leader_id(me_);
        request->set_prev_log_index(prev_log_index);
        request->set_perv_log_term(prev_log_term);
        request->set_leader_commint(commit_index_);
        request->clear_entries();
        // 如果 follower 的日志不是快照中最后一条，则从日志数组中提取日志
        // 计算日志条目的起始位置
        int start_index = (prev_log_index != last_snapshot_include_index_)
                                  ? get_slices_index_by_log_index(prev_log_index) + 1
                                  : 0;

        // 添加日志条目
        for (int j = start_index; j < logs_.size(); ++j) {
            auto send_entry_ptr = request->add_entries();
            *send_entry_ptr = logs_[j];
        }

        auto last_log_index = get_last_log_index();
        if (request->prev_log_index() + request->entries_size() != last_log_index) {
            log.fatal("[func-raft::core::RaftCore::initiate_heartbeat()] request->prev_log_index(){} + len(request->entries_size()){} != last_log_index{}",
                      request->prev_log_index(), request->entries_size(), last_log_index);
            co_return;
        }

        auto reply = std::make_shared<raft_rpc_protoc::AppendEntriesReply>();
        reply->set_app_state(0);// 0 代表建立连接
        // 启动一个协程发送
        this->send_append_entries(i, request, reply, append_nums).start([this](auto&&) {});
    }
    last_heartbeat_reset_time_ = std::chrono::high_resolution_clock::now();
    log.verbose("End: async_simple::coro::Lazy<void> raft::core::RaftCore::initiate_heartbeat()");
}

async_simple::coro::Lazy<bool>
raft::core::RaftCore::send_append_entries(int server,
                                          std::shared_ptr<raft_rpc_protoc::AppendEntries>& request,
                                          std::shared_ptr<raft_rpc_protoc::AppendEntriesReply>& reply,
                                          std::shared_ptr<std::atomic<int>> append_nums) {
    log.verbose("Start: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");

    log.debug("[func-raft::core::RaftCore::send_append_entries] {} leader sends an append_entries RPC to node {}, args->entries_size(): {}",
              me_, server, request->entries_size());
    bool result = peers_[server]->append_entries(*request, reply.get());
    if (!result) {
        log.debug("[func-raft::core::RaftCore::send_append_entries] send request_vote failed");
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
        co_return false;
    }

    // 如果不处理，则返回成功
    if (reply->app_state() == 0) {
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
        co_return true;
    }


    // 回复的任期号大于当前任期，转为 follower
    if (reply->term() > current_term_) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = reply->term();
            voted_for_.clear();
            leader_me_ = request->leader_id();
        }

        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
        co_return true;
    } else if (reply->term() < current_term_) {
        log.debug("[func-raft::core::RaftCore::send_append_entries] Node: {}'s term{}<raft{}'s term{}\n",
                  server, reply->term(), me_, current_term_);
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
        co_return true;
    }

    if (status_ != Status::Leader) {
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
        co_return true;
    }

    if (reply->term() != current_term_) {
        log.fatal("[func-raft::core::RaftCore::send_append_entries] reply->term(){} != current_term_{}", reply->term(), current_term_);
        co_return false;
    }

    if (!reply->success()) {
        if (reply->update_next_index() != -100) {
            log.debug("[func-raft::core::RaftCore::send_append_entries] {} The returned log term is equal, but does not match, retracting nextIndex{}: {}",
                      me_, server, reply->update_next_index());

            std::lock_guard<std::mutex> lock(mutex_);
            next_index_[server] = reply->update_next_index();
        }
    } else {
        // 如果成功，则增加成功响应的节点数量
        ++*append_nums;

        {
            std::lock_guard<std::mutex> lock(mutex_);
            match_index_[server] = std::max(match_index_[server], request->prev_log_index() + request->entries_size());
            ++match_index_[server];
        }

        // 如果超过半数节点成功响应，则提交日志
        if (*append_nums >= peers_.size() / 2) {
            *append_nums = 0;
            if (request->entries_size() > 0 &&
                request->entries(request->entries_size() - 1).log_term() == current_term_) {

                std::lock_guard<std::mutex> lock(mutex_);
                commit_index_ = std::max(commit_index_, request->prev_log_index() + request->entries_size());
                log.info("The current term has a new log submission, and the commit_index of the updated leader has changed from {} to {}.",
                         commit_index_, request->prev_log_index() + request->entries_size());
            }
        }
    }

    log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_append_entries(int server,...)");
    co_return true;
}

grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,
                                                  const raft_rpc_protoc::AppendEntries* request,
                                                  raft_rpc_protoc::AppendEntriesReply* reply) {
    log.verbose("Start: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");

    reply->set_app_state(1);

    // 如果任期号小于当前任期号，拒绝请求
    if (request->term() < current_term_) {
        reply->set_success(false);
        reply->set_term(current_term_);
        reply->set_update_next_index(-100);
        log.verbose("End: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    // 如果大于当前任期，跟新任期，并转为身份为 follower
    if (request->term() > current_term_) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = request->term();
            voted_for_.clear();
            leader_me_ = request->leader_id();
        }
        log.verbose("End: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    if (request->term() != current_term_) {
        log.fatal("[func-raft::core::RaftCore::append_entries] {} Previously verified that request->term() {} == current_term_ {} is not equal here.",
                  me_, request->term(), current_term_);
        return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, "fatal error");
    }

    // 重置选举时间
    {
        std::lock_guard<std::mutex> lock(mutex_);
        last_election_reset_time_ = std::chrono::high_resolution_clock::now();
        status_ = Status::Follower;
    }

    // 检查前一个日志索引是否大于最后的日志索引，如果是则拒绝
    if (request->prev_log_index() > get_last_log_index()) {
        reply->set_success(false);
        reply->set_term(current_term_);
        reply->set_update_next_index(last_snapshot_include_index_ + 1);
        log.verbose("End: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    // 日志匹配，则处理日志条目
    if (is_log_match(request->prev_log_index(), request->perv_log_term())) {
        // 遍历日志条目，添加到日志中
        for (int i = 0; i < request->entries_size(); ++i) {
            auto log_entry_log = request->entries(i);
            // 大于最后日志索引，直接添加
            if (log_entry_log.log_index() > get_last_log_index()) {
                std::lock_guard<std::mutex> lock(mutex_);
                logs_.emplace_back(log_entry_log);
            } else {
                // 不匹配则更新
                if (logs_[get_slices_index_by_log_index(log_entry_log.log_index())].log_term() !=
                    log_entry_log.log_term()) {
                    std::lock_guard<std::mutex> lock(mutex_);
                    logs_[get_slices_index_by_log_index(log_entry_log.log_index())] = log_entry_log;
                }
            }
        }

        // 更新提交的索引
        if (request->leader_commint() > commit_index_) {
            std::lock_guard<std::mutex> lock(mutex_);
            commit_index_ = std::max(request->leader_commint(), get_last_log_index());
        }

        // 设置回复成功
        reply->set_success(true);
        reply->set_term(current_term_);
        log.verbose("End: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    // 日志不匹配
    reply->set_update_next_index(request->prev_log_index());
    for (int i = request->prev_log_index(); i >= last_snapshot_include_index_; --i) {
        if (get_log_term_by_log_index(i) != get_log_term_by_log_index(request->prev_log_index())) {
            reply->set_update_next_index(i + 1);
            break;
        }
    }
    reply->set_success(false);
    reply->set_term(current_term_);

    log.verbose("End: grpc::Status raft::core::RaftCore::append_entries(grpc::ServerContext* context,...)");
    return grpc::Status::OK;
}

async_simple::coro::Lazy<bool>
raft::core::RaftCore::send_install_snapshot(int server,
                                            std::shared_ptr<raft_rpc_protoc::InstallSnapshot>& request,
                                            std::shared_ptr<raft_rpc_protoc::InstallSnapshotReply>& reply) {
    log.verbose("Start: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_install_snapshot(int server,...)");

    bool result = peers_[server]->install_snapshot(*request, reply.get());

    if (!result) {
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_install_snapshot(int server,...)");
        co_return false;
    }

    if (status_ != Status::Leader || current_term_ != request->term()) {
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_install_snapshot(int server,...)");
        co_return false;
    }

    // 如果回复大于当前任期号，转为跟随着
    if (reply->term() > current_term_) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = reply->term();
            voted_for_.clear();
            leader_me_ = request->leader_id();
        }
        persistence();
        {
            std::lock_guard<std::mutex> lock(mutex_);
            last_election_reset_time_ = std::chrono::high_resolution_clock::now();
        }
        log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_install_snapshot(int server,...)");
        co_return true;
    }

    // 更新匹配索引和下一个索引
    {
        std::lock_guard<std::mutex> lock(mutex_);
        match_index_[server] = request->last_snapshot_include_index();
        ++match_index_[server];
    }
    log.verbose("End: async_simple::coro::Lazy<bool> raft::core::RaftCore::send_install_snapshot(int server,...)");
    co_return true;
}

grpc::Status raft::core::RaftCore::install_snapshot(grpc::ServerContext* context,
                                                    const raft_rpc_protoc::InstallSnapshot* request,
                                                    raft_rpc_protoc::InstallSnapshotReply* reply) {
    log.verbose("Start: grpc::Status raft::core::RaftCore::install_snapshot(grpc::ServerContext* context,...)");

    // 如果请求的任期号小于当前任期，拒绝请求
    if (request->term() < current_term_) {
        reply->set_term(current_term_);
        log.verbose("End: grpc::Status raft::core::RaftCore::install_snapshot(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    // 如果请求的任期号大于当前任期，更新当前任期，转换为跟随者，并重置投票状态
    if (request->term() > current_term_) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            status_ = Status::Follower;
            current_term_ = request->term();
            voted_for_.clear();
            leader_me_ = request->leader_id();
        }
        persistence();
    }

    {
        std::lock_guard<std::mutex> lock(mutex_);
        status_ = Status::Follower;
        last_election_reset_time_ = std::chrono::high_resolution_clock::now();
    }
    // 检查快照是否过期，过期则拒绝快照
    if (request->last_snapshot_include_index() <= last_snapshot_include_index_) {
        log.verbose("End: grpc::Status raft::core::RaftCore::install_snapshot(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }
    // 获取最后一个日志索引
    int last_log_index = get_last_log_index();
    {
        std::lock_guard<std::mutex> lock(mutex_);

        if (last_log_index > request->last_snapshot_include_index()) {
            // 如果最后日志索引大于快照包含的索引，截断日志
            logs_.erase(logs_.begin(), logs_.begin() + get_slices_index_by_log_index(request->last_snapshot_include_index()) + 1);
        } else {
            // 如果快照包含的索引大于等于最后日志索引，清空日志
            logs_.clear();
        }
        // 更新快照信息
        last_snapshot_include_index_ = request->last_snapshot_include_index();
        last_snapshot_include_term_ = request->last_snapshot_include_term();

        commit_index_ = std::max(commit_index_, last_snapshot_include_index_);
        last_applied_ = std::max(last_applied_, last_snapshot_include_index_);
    }

    reply->set_term(current_term_);

    // 应用快照到状态机
    ApplyMsg msg;
    msg.set_snapshot_valid(true);
    msg.set_snapshot(request->data());
    msg.set_snapshot_index(last_snapshot_include_index_);
    msg.set_snapshot_term(last_snapshot_include_term_);

    // 创建一个线程，用于快照的状态机
    // 使用 lambda 表达式捕获参数
    std::thread t([this, &msg]() { this->apply_log(msg); });
    t.detach();

    // 持久化快照和日志
    persistence_->save(serialize_data(), request->data());
    log.verbose("End: grpc::Status raft::core::RaftCore::install_snapshot(grpc::ServerContext* context,...)");
    return grpc::Status::OK;
}

grpc::Status raft::core::RaftCore::execute_command(grpc::ServerContext* context,
                                                   const middleware_rpc_proto::ExecuteCommand* request,
                                                   middleware_rpc_proto::ExecuteCommandReply* reply) {
    log.verbose("Start: grpc::Status raft::core::RaftCore::execute_command(grpc::ServerContext* context,...)");

    if (status_ != Status::Leader) {
        reply->set_is_success(false);
        reply->set_result(leader_me_);
        log.verbose("End: grpc::Status raft::core::RaftCore::execute_command(grpc::ServerContext* context,...)");
        return grpc::Status::OK;
    }

    raft_rpc_protoc::LogEntry log_entry;
    log_entry.set_command(request->apply_msg());
    log_entry.set_log_index(get_last_log_index() + 1);
    log_entry.set_log_term(current_term_);
    logs_.emplace_back(log_entry);

    log.info("[func-raft::core::RaftCore::execute_command] {} received the command.",
             me_);

    // 启动一个协程触发日志
    initiate_heartbeat().start([](auto&&) {});

    reply->set_is_success(true);
    reply->set_result("execute_command success.");
    log.verbose("End: grpc::Status raft::core::RaftCore::execute_command(grpc::ServerContext* context,...)");
    return grpc::Status::OK;
}